package com.atguigu.windowFunction;

import com.atguigu.bean.LoginEvent;
import com.atguigu.utils.MyUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;

/**
 * Created by RUI on 2021/6/18 16:11
 *
 * 同一用户（可以是不同IP）在2秒之内连续两次登录失败，就认为存在恶意登录的风险，输出相关的信息进行报警提示
 */
public class High_prj_Login {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 创建WatermarkStrategy
        WatermarkStrategy<LoginEvent> wms = WatermarkStrategy
                .<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                .withTimestampAssigner((element, recordTimestamp) -> element.getEventTime());
        env
                .readTextFile("input/LoginLog.csv")
                .map(line -> {
                    String[] data = line.split(",");
                    return new LoginEvent(Long.valueOf(data[0]),
                            data[1],
                            data[2],
                            Long.parseLong(data[3]) * 1000L);
                })
                .assignTimestampsAndWatermarks(wms)
                .keyBy(LoginEvent::getUserId)
                .process(new KeyedProcessFunction<Long, LoginEvent, String>() {

                    private ListState<LoginEvent> logState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        logState = getRuntimeContext().getListState(new ListStateDescriptor<LoginEvent>("logState", LoginEvent.class));
                    }

                    @Override
                    public void processElement(LoginEvent value, Context ctx, Collector<String> out) throws Exception {

                        if ("fail".equals(value.getEventType())){
                            logState.add(value);
                            List<LoginEvent> loginEvents = MyUtils.toList(logState.get());
                            if (loginEvents.size()==2){
                                out.collect(value.getUserId()+" 正在恶意登录");
                                logState.clear();
                                logState.add(loginEvents.get(1));
                            }
                        }else {
                            logState.clear();
                        }

                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
